package com.example.hbase;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.regionserver.KeyPrefixRegionSplitPolicy;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Assert;

public class HBaseImportEx {

    static Configuration hbaseConfig;

    // 执行插入操作的线程数
    public static final int THREAD_COUNT = 50;

    // 一次批量插入操作的消息数
    public static final int BATCH_INSERT_NUM = 9999999;

    // 表名
    public static String tableName = "water2";

    // 列族名
    private String columnFamily = "jx";

    private Admin admin;

    private Connection connection;

    private Configuration configuration;

    /**
     * 获取连接
     */
    private void init() {
        try {
            configuration = HBaseConfiguration.create();
            configuration.set("hbase.zookeeper.property.clientPort", "2181");
            configuration.set("hbase.zookeeper.quorum", "cdh0");
            configuration.set("hbase.client.pause", "200");
            configuration.set("hbase.ipc.client.tcpnodelay", "true");
            configuration.set("hbase.client.write.buffer", "100");
            System.out.println("---------创建连接用时测试----------");
            long start = System.currentTimeMillis();
            connection = ConnectionFactory.createConnection(configuration);
            long stop = System.currentTimeMillis();
            System.out.println("创建连接用时共耗时：" + (stop - start) * 1.0 / 1000 + "s");
            admin = connection.getAdmin();
        } catch (Throwable e) {
            e.printStackTrace();
        }
    }

    /**
     * 创建表
     */
    private void createTable() {
        Table table = null;
        try {
            if (admin.tableExists(TableName.valueOf(tableName))) {
                Assert.fail();
            }
            HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
            tableDescriptor.addFamily(new HColumnDescriptor(columnFamily));
            tableDescriptor.setValue(HTableDescriptor.SPLIT_POLICY, KeyPrefixRegionSplitPolicy.class .getName());
            tableDescriptor.setValue("prefix_split_key_policy.prefix_length", "1");
            admin.createTable(tableDescriptor);
            System.out.println("表已创建");
        } catch (Throwable e) {
            e.printStackTrace();
            Assert.fail();
        }
    }

    /**
     * 删除表
     */
    private void deleteTable() {
        try {
            if (!admin.tableExists(TableName.valueOf(tableName))) {
                Assert.fail();
                return;
            }
            admin.disableTable(TableName.valueOf(tableName));
            admin.deleteTable(TableName.valueOf(tableName));
            System.out.println("表已删除");
        } catch (Throwable e) {
            e.printStackTrace();
            Assert.fail();
        }
    }

    /**
     * Insert Test single thread
     */
    public static void SingleThreadInsert() throws IOException {
        System.out.println("---------开始SingleThreadInsert测试----------");
        long start = System.currentTimeMillis();
        HTable table = null;
        table = new HTable(hbaseConfig, tableName);
        table.setAutoFlush(false, true);
        table.setWriteBufferSize(36 * 1024 * 1024);
        System.out.print("Auto Flush : " + table.isAutoFlush());
        // 构造测试数据
        List<Put> list = new ArrayList<Put>();
        byte[] buffer = new byte[350];
        Random rand = new Random();
        for (int i = 0; i < BATCH_INSERT_NUM; i++) {
            Put put = new Put(String.format("row %d", i).getBytes());
            rand.nextBytes(buffer);
            put.add("cf1".getBytes(), null, buffer);
            put.setWriteToWAL(false);
            list.add(put);
            System.out.println(i);
            if (i % 100 == 0) {
                table.put(list);
                table.flushCommits();
                list.clear();
            }
        }
        long stop = System.currentTimeMillis();
        System.out.println("插入数据：" + BATCH_INSERT_NUM + "共耗时：" + (stop - start) * 1.0 / 1000 + "s");
        System.out.println("---------结束SingleThreadInsert测试----------");
    }

    /*
     * 多线程环境下线程插入函数
     */
    public void InsertProcess() throws IOException {
        System.out.println("线程:" + Thread.currentThread().getId() + "开始插入数据：");
        long start = System.currentTimeMillis();
        // HTableInterface table = null;
        HTable table = null;
        table = (HTable) connection.getTable(TableName.valueOf(tableName));
        System.out.println("连接的哈希 :" + connection.hashCode());
        System.out.println("表的哈希 :" + table.hashCode());
        table.setAutoFlush(false, true);
        table.setWriteBufferSize(24 * 1024 * 1024);
        // 构造测试数据
        List<Put> list = new ArrayList<Put>();
        Random rand = new Random();
        char c=(char)(int)(Math.random()*26+97);
        for (int i = 0; i < BATCH_INSERT_NUM; i++) {
            long perfix = Long.MAX_VALUE - System.nanoTime();

            Put put = new Put(Bytes.toBytes(c+ "_" + perfix ));
            String id = Thread.currentThread().getId() + "-" + i;
            put.add(Bytes.toBytes(columnFamily), Bytes.toBytes("id"), Bytes.toBytes(id));
            put.add(Bytes.toBytes(columnFamily), Bytes.toBytes("consumption"),
                    Bytes.toBytes(String.valueOf(rand.nextInt(3))));
            put.add(Bytes.toBytes(columnFamily), Bytes.toBytes("jzqid"), Bytes.toBytes(id));
            put.add(Bytes.toBytes(columnFamily), Bytes.toBytes("state"),
                    Bytes.toBytes(String.valueOf(rand.nextInt(5))));
            put.setWriteToWAL(false);
            list.add(put);
            if (i % 100 == 0) {
                table.put(list);
                table.flushCommits();
                System.out.println(list.size());
                list.clear();
            }
        }
        if (table != null) {
            try {
                table.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        long stop = System.currentTimeMillis();
        // System.out.println("WAL="+wal+",autoFlush="+autoFlush+",buffer="+writeBuffer+",count="+count);
        System.out.println("线程:" + Thread.currentThread().getId() + "插入数据：" + BATCH_INSERT_NUM
                + "共耗时：" + (stop - start) * 1.0 / 1000 + "s");
    }

    /*
     * Mutil thread insert test
     */
    private void batchInsert() throws IOException, InterruptedException {
        System.out.println("---------开始MultThreadInsert测试----------");
        long start = System.currentTimeMillis();
        Thread[] threads = new Thread[THREAD_COUNT];
        for (int i = 0; i < threads.length; i++) {
            threads[i] = new ImportThread();
            threads[i].start();
        }
        for (int j = 0; j < threads.length; j++) {
            (threads[j]).join();
        }
        long stop = System.currentTimeMillis();

        System.out.println("MultThreadInsert：" + THREAD_COUNT * BATCH_INSERT_NUM + "共耗时："
                + (stop - start) * 1.0 / 1000 + "s");
        System.out.println("---------结束MultThreadInsert测试----------");
    }

    /**
     * @param args
     */
    public static void main(String[] args) throws Exception {
        HBaseImportEx ex = new HBaseImportEx();
        ex.init();
//        ex.deleteTable();
        ex.createTable();
//        ex.batchInsert();
    }

    public class ImportThread extends Thread {
        public void run() {
            try {
                InsertProcess();
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                System.gc();
            }
        }
    }
}
